package com.lry;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * 提议者
 */
@Slf4j
public class Proposer {

    private List<Acceptor> acceptors;
    private Integer round;
    private Integer value;
    private Integer excludeIndex;//用于测试机器宕机

    public Proposer(Integer excludeIndex,Integer round,Integer value){
        acceptors = new ArrayList<>();
        this.round = round;
        this.value = value;
        this.excludeIndex = excludeIndex;
    }

    public void registerAcceptors(Acceptor... inAcceptors){
        for (Acceptor inAcceptor : inAcceptors) {
            if(null==inAcceptor){
                continue;
            }
            acceptors.add(inAcceptor);
        }
    }

    public void prepareRequest(){
        List<CompletableFuture<Phase1Resp>> futures = new ArrayList<>();

        for (int i =0;i<acceptors.size();i++) {
            if(i==excludeIndex){
                continue;
            }
            Acceptor acceptor = acceptors.get(i);
            CompletableFuture<Phase1Resp> future = CompletableFuture.supplyAsync(()-> acceptor.prepareRequest(round));
            futures.add(future);
        }

        int successNum = 0;
        Integer valueRoundMax = Integer.MIN_VALUE;
        Integer valueMax = Integer.MIN_VALUE;
        for (CompletableFuture<Phase1Resp> future : futures) {
            try {
                Phase1Resp resp = future.get();
                if(null==resp){
                    continue;
                }
                if(!RespEnums.SUCCESS.getResp().equals(resp.getRespCode())){
                    continue;
                }
                successNum++;
                if(resp.getLastRound()!=null && resp.getLastRound()>round){
                    //返回的response中如果有last_round 大于round则直接退出
                    log.info("prepareRequest,返回的response中如果有last_round 大于round则直接退出,lastRound={},round={},value={}",resp.getLastRound(),round,value);
                    return;
                }
                //从所有response中选择value_round 最大的那个value
                if(resp.getValueRound()!=null&&resp.getValueRound()>valueRoundMax){
                    valueRoundMax = resp.getValueRound();
                    valueMax = resp.getValue();
                }

            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }
        //如果没有得到一个多数派集合的确认，则直接退出
        if(successNum<=acceptors.size()/2){
            log.info("prepareRequest，一阶段没有得到一个多数派集合的确认，直接退出，round={},value={}",round,value);
            return;
        }

        //如果所有response中的value都是空，则选择自己提议的value
        if(valueRoundMax == Integer.MIN_VALUE && valueMax == Integer.MIN_VALUE){
            acceptRequest(value);
        }else{
            acceptRequest(valueMax);
        }
    }

    private void acceptRequest(Integer curValue){
        List<CompletableFuture<Integer>> futures = new ArrayList<>();
        for (int i =0;i<acceptors.size();i++) {
            if(i==excludeIndex){
                continue;
            }
            Acceptor acceptor = acceptors.get(i);
            CompletableFuture <Integer> future = CompletableFuture.supplyAsync(()-> acceptor.acceptRequest(round,curValue));
            futures.add(future);
        }
        int successNum = 0;
//        只要获得一个多数派的写入确认的支持，则表明其value通过提案
        for (CompletableFuture<Integer> future : futures) {
            try {
                Integer respCode = future.get();
                if(RespEnums.SUCCESS.getResp().equals(respCode)){
                    successNum++;
                }
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }

        if(successNum>acceptors.size()/2){
            log.info("acceptRequest,响应通过提案，round={},value={}",round,curValue);
            return;
        }

        log.info("acceptRequest,响应不通过提案，增加round继续paxos 一阶段，round={},value={}",round,curValue);
        //增加round，继续paxos 一阶段
        round++;
        prepareRequest();
    }


}
